package com.lry.index;

import com.lry.DefaultMessageStore;
import com.lry.DispatchRequest;
import com.lry.UtilAll;
import com.lry.config.StorePathConfigHelper;
import com.lry.message.MessageConst;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

@Slf4j
public class IndexService {
    private static final int MAX_TRY_IDX_CREATE = 3;
    private final int hashSlotNum;
    private final int indexNum;
    private final String storePath;
    private final ArrayList<IndexFile> indexFileList = new ArrayList<IndexFile>();
    private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    private final DefaultMessageStore defaultMessageStore;

    public IndexService(final DefaultMessageStore store) {
        this.defaultMessageStore = store;
        this.hashSlotNum = store.getMessageStoreConfig().getMaxHashSlotNum();
        this.indexNum = store.getMessageStoreConfig().getMaxIndexNum();
        this.storePath =
                StorePathConfigHelper.getStorePathIndex(store.getMessageStoreConfig().getStorePathRootDir());
    }

    public void destroy() {
        try {
            this.readWriteLock.writeLock().lock();
            for (IndexFile f : this.indexFileList) {
//                f.destroy(1000 * 3);
            }
            this.indexFileList.clear();
        } catch (Exception e) {
            log.error("destroy exception", e);
        } finally {
            this.readWriteLock.writeLock().unlock();
        }
    }


    public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) {
        //匹配到的所有偏移量的集合
        List<Long> phyOffsets = new ArrayList<Long>(maxNum);

        long indexLastUpdateTimestamp = 0;
        long indexLastUpdatePhyoffset = 0;
        maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());

        try {
            this.readWriteLock.readLock().lock();

            if (!this.indexFileList.isEmpty()) {
                for (int i = this.indexFileList.size(); i > 0; i--) {
                    IndexFile f = this.indexFileList.get(i - 1);
                    boolean lastFile = i == this.indexFileList.size();

                    if (lastFile) {
                        indexLastUpdateTimestamp = f.getEndTimestamp();
                        indexLastUpdatePhyoffset = f.getEndPhyOffset();
                    }

                    if (f.isTimeMatched(begin, end)) {
                        f.selectPhyOffset(phyOffsets, buildKey(topic, key), maxNum, begin, end, lastFile);
                    }

                    if (f.getBeginTimestamp() < begin) {
                        break;
                    }

                    if (phyOffsets.size() >= maxNum) {
                        break;
                    }
                }
            }
        } catch (Exception e) {
            log.error("queryMsg exception", e);
        } finally {
            this.readWriteLock.readLock().unlock();
        }

        return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset);
    }

    private String buildKey(final String topic, final String key) {
        return topic + "#" + key;
    }

    public void buildIndex(DispatchRequest req) {
        IndexFile indexFile = retryGetAndCreateIndexFile();
        if(null==indexFile){
            log.error("build index error, stop building index");
            return;
        }
        long endPhyOffset = indexFile.getEndPhyOffset();
        String topic = req.getTopic();
        String keys = req.getKeys();
        if (req.getCommitLogOffset() < endPhyOffset) {
            return;
        }


        if(StringUtils.isNotEmpty(keys)){
            String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
            for (String key : keyset) {
                if(StringUtils.isNotEmpty(key)) {
                    indexFile = putKey(indexFile, req, buildKey(topic, key));
                    if (indexFile == null) {
                        log.error("putKey error commitlog {}", req.getCommitLogOffset());
                        return;
                    }
                }
            }
        }
    }

    private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) {
        for (boolean ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok; ) {
            log.warn("Index file [" + indexFile.getFileName() + "] is full, trying to create another one");

            indexFile = retryGetAndCreateIndexFile();
            if (null == indexFile) {
                return null;
            }

            ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp());
        }

        return indexFile;
    }

    private IndexFile retryGetAndCreateIndexFile() {
        IndexFile indexFile = null;

        for (int times = 0; null == indexFile && times < MAX_TRY_IDX_CREATE; times++) {
            indexFile = this.getAndCreateLastIndexFile();
            if (null != indexFile){
                break;
            }
            try {
                log.info("Tried to create index file " + times + " times");
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                log.error("Interrupted", e);
            }
        }

        return indexFile;
    }

    private IndexFile getAndCreateLastIndexFile() {
        IndexFile indexFile = null;
        IndexFile prevIndexFile = null;
        long lastUpdateEndPhyOffset = 0;
        long lastUpdateIndexTimestamp = 0;

        this.readWriteLock.readLock().lock();
        if(!this.indexFileList.isEmpty()){
            //拿到最后一个index file
            IndexFile tmp = this.indexFileList.get(this.indexFileList.size() - 1);
            if(!tmp.isWriteFull()){
                indexFile = tmp;
            }else{
                //上个index file的一些信息
                lastUpdateEndPhyOffset = tmp.getEndPhyOffset();
                lastUpdateIndexTimestamp = tmp.getEndTimestamp();
                prevIndexFile = tmp;
            }
        }
        this.readWriteLock.readLock().unlock();

        if(null!=indexFile){
            return indexFile;
        }

        //上个文件已满，则新建一个index file
        String fileName =
                this.storePath + File.separator
                        + UtilAll.timeMillisToHumanString(System.currentTimeMillis());
        //新建一个index file
        try {
            indexFile = new IndexFile(fileName,this.hashSlotNum,this.indexNum,lastUpdateEndPhyOffset,lastUpdateIndexTimestamp);
            this.readWriteLock.writeLock().lock();
            this.indexFileList.add(indexFile);
        } catch (IOException e) {
            log.error("getLastIndexFile exception ", e);
        }finally {
            this.readWriteLock.writeLock().unlock();
        }

        if (indexFile != null) {
            final IndexFile flushThisFile = prevIndexFile;
            Thread flushThread = new Thread(()->{
                //上个index file 写满了，刷到磁盘
                IndexService.this.flush(flushThisFile);
            }, "FlushIndexFileThread");

            flushThread.setDaemon(true);
            flushThread.start();
        }

        return indexFile;
    }

    public void flush(final IndexFile f) {
        if (null == f){
            return;
        }
        f.flush();
    }


    public void shutdown() {
    }

    public boolean load() {
        File dir = new File(this.storePath);
        File[] files = dir.listFiles();
        if (files != null) {
            // ascending order
            Arrays.sort(files);
            for (File file : files) {
                try {
                    IndexFile f = new IndexFile(file.getPath(), this.hashSlotNum, this.indexNum, 0, 0);
                    f.load();

                    log.info("load index file OK, " + f.getFileName());
                    this.indexFileList.add(f);
                } catch (IOException e) {
                    log.error("load file {} error", file, e);
                    return false;
                } catch (NumberFormatException e) {
                    log.error("load file {} error", file, e);
                }
            }
        }

        return true;
    }
}
